
#include <compressor.hpp>
#include "io_service_pool_impl.hpp"
#include "client_environment.hpp"

xbase::client_environment_ptr env;

boost::mutex m;
std::size_t pending_ops;

bool sync_mode = false;

const std::string default_table_name_prefix("test_table_mutator-");
std::string table_name_prefix;

void handler(const char *op,
	     const std::string &table_name,
	     const boost::system::error_code &ec)
{
	std::cout << "table \"" << table_name << "\" "
		  << op << ", ec = "
		  << ec << " - " << ec.message()
		  << std::endl;
	boost::mutex::scoped_lock lock(m);
	--pending_ops;
}

void fill_creation(xbase::table_creation &creation)
{
	xbase::table_creation::group_options g_options;
	g_options.m_compress_method = xbase::compressor::NONE;
	g_options.m_cache_rate = 30;
	g_options.m_block_size = 64 * 1024;
	g_options.m_max_num_sstable = 9;
	g_options.m_max_size_sstable = 128 * 1024 * 1024;
	g_options.m_max_size_memtable = 32 * 1024 * 1024;

	xbase::table_creation::family_options f_options;
	f_options.m_max_num_version = 3;
	f_options.m_max_ttl = 3 * 24 * 60 * 60;
	f_options.m_dead_time = 0;
	
	creation.add_group("content", g_options);
	creation.add_group("attr", g_options);
	creation.add_group("info", g_options);
	
	creation.add_family("contents", "content", f_options);
	creation.add_family("attrs", "attr", f_options);
	creation.add_family("detail", "info", f_options);
	creation.add_family("anchor", "info", f_options);
}

void async_delete_table(const std::string &table_name)
{
	xbase::table_deletion deletion(table_name);
	if(sync_mode)
	{
		xbase::table_mutator::sync_handler_type sync_handler;
		env->async_apply(deletion, sync_handler.make_handler());
		++pending_ops;
		sync_handler.wait();
		handler("deleted", table_name, sync_handler.get_error_code());
	}
	else
	{
		env->async_apply(deletion, boost::bind(handler, "deleted", table_name, _1));
		boost::mutex::scoped_lock lock(m);
		++pending_ops;
	}
}

void async_create_table(const std::string &table_name)
{
	xbase::table_creation creation(table_name);
	fill_creation(creation);
	if(sync_mode)
	{
		xbase::table_mutator::sync_handler_type sync_handler;
		env->async_apply(creation, sync_handler.make_handler());
		++pending_ops;
		sync_handler.wait();
		handler("created", table_name, sync_handler.get_error_code());
	}
	else
	{
		env->async_apply(creation, boost::bind(handler, "created", table_name, _1));
		boost::mutex::scoped_lock lock(m);
		++pending_ops;
	}
}

void test_async_mutation_on_table(const char *op)
{
	for(std::size_t i=0; i<100; ++i)
	{
		if(op == std::string("create"))
		{
			async_create_table(table_name_prefix + boost::lexical_cast<std::string>(i));
		}
		else if(op == std::string("delete"))
		{
			async_delete_table(table_name_prefix + boost::lexical_cast<std::string>(i));
		}
	}
}

int main(int argc, char **argv)
{
	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	if(argc < 3 || argc > 5)
	{
		std::cout << "test_table_mutator <master ip> <master port> [sync_mode] [table_prefix]" << std::endl;
		return 0;
	}
	
	boost::asio::ip::tcp::endpoint mep(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);

	sync_mode = false;
	if(argc >= 4)
		if(argv[3] == std::string("true"))
			sync_mode = true;
			
	if(argc >= 5)
		table_name_prefix = argv[4];
	else
		table_name_prefix = default_table_name_prefix;

	xbase::async_module_stats::enable();
	{

		xbase::client_options_ptr options(new xbase::client_options);
		xbase::io_service_pool_ptr pool(new xbase::impl::single_io_service());
		env.reset(new xbase::client_environment(options, pool, mep));

		std::cout << "testing create table ... " << std::endl;
		test_async_mutation_on_table("create");
		boost::mutex::scoped_lock lock(m);
		while(pending_ops > 0)
		{
			lock.unlock();

			xbase::async_module_stats::print(std::cout);
			std::cout << std::endl;
			std::cout << "wait pending ops ..." << std::endl;
			sleep(1);
		
			lock.lock();
		}
		lock.unlock();
		std::cout << "all tables created" << std::endl;
		xbase::async_module_stats::print(std::cout);
		std::cout << std::endl;

		sleep(3);

		std::cout << "testing delete table ... " << std::endl;
		test_async_mutation_on_table("delete");
		lock.lock();
		while(pending_ops > 0)
		{
			lock.unlock();

			std::cout << "wait pending ops ..." << std::endl;
			xbase::async_module_stats::print(std::cout);
			std::cout << std::endl;
			sleep(1);
		
			lock.lock();
		}
		lock.unlock();
		std::cout << "all tables deleted" << std::endl;
		xbase::async_module_stats::print(std::cout);
		std::cout << std::endl;

		sleep(3);

		env->stop();
		env->join();
		env.reset();
	}

	xbase::async_module_stats::print(std::cout);
	std::cout << std::endl;

	return 0;
}
